package com.example.kafkaboot.controller;

import com.example.kafkaboot.model.User;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;
import java.util.Set;

/**
 * @author cxx
 * @create 2020-03-30 15:56
 **/
@RestController
public class UserController {

    @Autowired
    private KafkaTemplate<Object, Object> template;

    @RequestMapping("/to/{message}")
    public String toMessage(@PathVariable String message){
        template.executeInTransaction (kafkaTemplate  -> {
            StringUtils.commaDelimitedListToSet(message).stream()
                    .map(s -> new User (s))
                    .forEach(user -> {
                        kafkaTemplate.send("test.user", user);
                    });
            return null;
        });
        return message;
    }

    @KafkaListener(id = "userGroup1", topics = "test.user")
    public void kafkaListener(List<User> list){
        try {
            for (User user : list) {
                System.out.println ("userGroup1 = "+ user.getName ());
            }
        } finally {
        }
    }

}
